package com.gaussdb.process;


import com.gaussdb.config.MyConfig;
import com.gaussdb.source.GaussDBSourceFunction;
import com.gaussdb.source.GaussDBSqlSource;
import com.gaussdb.tansform.SourceMapFunction;
import com.gaussdb.tool.DataSqlUtil;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class GaussToDorisAuto {
    // 程序的主入口点
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1
        env.setParallelism(1);

        // 构建GaussDB数据源函数
        GaussDBSourceFunction<String> gaussSource = GaussDBSqlSource.<String>builder()
                .hostname(MyConfig.hostname)        // 服务器地址
                .port(8001)                         // 端口
                .username(MyConfig.gauss_username)  // 用户
                .password(MyConfig.gauss_password)  // 密码
                .database(MyConfig.gauss_database)  // 库名
                .tableList(MyConfig.white_table_list)    // 单表：public.tablename 全表：public.* 多表：public.tableA,public.tableB
                .slotName(MyConfig.gauss_replication_slot)  // 复制槽
                .replicationMode(MyConfig.gauss_replication_mode)  // 逻辑复制槽模式： 1 创建逻辑复制槽 ，2 使用逻辑复制槽 ，3 删除逻辑复制槽
                .build();

        // 添加数据源
        DataStreamSource<String> dataStreamSource = env.addSource(gaussSource);
        dataStreamSource.print();

        // 对数据流进行转换处理
        SingleOutputStreamOperator<Tuple5<String, String, String, String, String>> dataStreamMap = dataStreamSource.map(new SourceMapFunction());

        // 打印转换后的数据流
        dataStreamMap.print();

        // 添加自定义SinkFunction
        dataStreamMap.addSink(new RichSinkFunction<Tuple5<String, String, String, String, String>>() {
            Connection conn = null;
            PreparedStatement preparedStatement = null;

            // 初始化资源
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                Class.forName("com.mysql.jdbc.Driver");
                // 建立数据库连接
                conn = DriverManager.getConnection(MyConfig.doris_jdbc, MyConfig.doris_username, MyConfig.doris_password);
            }

            // 释放资源
            @Override
            public void close() throws Exception {
                if (preparedStatement != null) {
                    preparedStatement.close();
                }
                if (conn != null) {
                    conn.close();
                }
                super.close();
            }

            // 数据处理逻辑
            @Override
            public void invoke(Tuple5<String, String, String, String, String> value, Context context) throws Exception {
                // 表名
                String tableName = value.f0;

                // 操作类型
                String opType = value.f1;

                // 字段内容
                String[] columnsVal = value.f4.split("&&&&");

                // 字段类型
                String[] columnsType = value.f3.split("&&&&");

                // 调用工具类插入数据
                try {
                    if (opType.equals("DELETE")) {
                        // 不删除数据，修改操作类型字段
                        DataSqlUtil.updateData(conn, tableName, opType, columnsType, columnsVal);
                    } else {
                        DataSqlUtil.insertData(conn, tableName, opType, columnsType, columnsVal);
                    }
                } catch (SQLException e) {
                    // 捕获SQL异常，并记录详细的异常信息
                    System.err.println("SQL error occurred: " + e.getMessage());
                    e.printStackTrace();
                } catch (Exception e) {
                    // 捕获其他异常，并记录详细的异常信息
                    System.err.println("An error occurred: " + e.getMessage());
                    e.printStackTrace();
                }

            }
        });

        // 执行环境
        env.execute();
    }
}


